package com.xiaofan.scala

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer011, FlinkKafkaProducer011}

import java.util.Properties

object StreamingKafkaSink_B0004 {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.enableCheckpointing(1000)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

    val text: DataStream[String] = env.socketTextStream("192.168.1.27", 9999)

    val topic = "flink_kafka_source_B0004"

    val properties = new Properties();
    properties.setProperty("bootstrap.servers", "192.168.1.23:9091,192.168.1.24:9091,192.168.1.25:9091")
    properties.setProperty("transaction.timeout.ms", String.valueOf(60000 * 15))

    val myProducer: FlinkKafkaProducer011[String] = new FlinkKafkaProducer011[String](topic, new KeyedSerializationSchemaWrapper[String](new SimpleStringSchema()), properties, FlinkKafkaProducer011.Semantic.EXACTLY_ONCE)
    text.addSink(myProducer).name("kafkaSink")

    env.execute("StreamingKafkaSink_B0004")
  }
}
